package com.at.bigdata.spark.core.hc

import com.at.bigdata.spark.core.hc.aviator.MultiplicationFunc
import com.googlecode.aviator.AviatorEvaluator
import org.apache.commons.lang3.time.DateUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util
import java.util.Date
import scala.collection.{JavaConverters, mutable}
import scala.collection.JavaConverters._
import scala.collection.convert.ImplicitConversions.`map AsJavaMap`
import scala.util.Random

/**
 *
 * @author cdhuangchao3
 * @date 2023/3/6 8:40 PM
 */
object Spark01_Aviator {

  def test1(sc: SparkContext) = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4))

    // aviator执行表达式
    val rdd1 = rdd.filter {
      n => {
        val l = AviatorEvaluator.execute("1+1")
        l == n
      }
    }
    println("执行表达式： " + rdd1.collect().mkString(","))
  }

  def test2(sc: SparkContext) = {
    val rdd = sc.makeRDD(List(
      ("a", 1), ("b", 2), ("c", 3)
    ))
    // 变量使用
    //    Map<String, Int> s = null;
    val rdd1 = rdd.map {
      t => {
        val map: Map[String, Any] = Map("a" -> 1, "b" -> 2, "c" -> 3)
        //        val jMap = new java.util.HashMap[String, Integer](JavaConverters.mapAsJavaMap(map))
        val env = map.asInstanceOf[Map[String, Object]]
//        val env = map.mapValues(_.asInstanceOf[Object])
        val str = AviatorEvaluator.execute(" 'hello' + a", env.asJava)
        //val value = AviatorEvaluator.getInstance().getClass.getMethod("execute", classOf[Array[AnyRef]]).invoke(AviatorEvaluator.getInstance(), Array(" 'hello' + a", map.asJava))
        (t._1, str + t._2.toString)
      }
    }
    println("test2 " + rdd1.collect().mkString(","))

  }

  def test0(sc: SparkContext) = {
    val env = Map("a" -> 1, "b" -> 2, "c" -> 3).mapValues(_.asInstanceOf[Object])
    val result = AviatorEvaluator.execute("a + b + c", env.asJava)
    println("test0 " + result) // 输出 6
  }

  def test3(sc: SparkContext) = {
    val rdd = sc.makeRDD(List(1, 2, 3, 4))
    AviatorEvaluator.addFunction(new MultiplicationFunc())

    val list = List(
      AviatorEvaluator.execute("string.substring('habcdello', 1, 3)"),
      AviatorEvaluator.execute("string.contains(\"abc\", 'ab')"),
      AviatorEvaluator.execute("string.contains('abc', string.substring('habcdello', 1, 3))"),
      AviatorEvaluator.execute("string_to_date('2018-09-11', 'yyyyMMdd')"),
      AviatorEvaluator.exec("date_to_string(dd, 'yyyyMMdd')", new Date().asInstanceOf[Object]),
      AviatorEvaluator.execute("multiplication(10,multiplication(5,4))")
    )

    val rdd1 = rdd.map {
      i => (i, list(Random.nextInt(list.size)).toString)
    }
    println("test3" + rdd1.collect().mkString(","))
  }

  def test4(sc: SparkContext) = {
    val javaMap = Map("a" -> 100, "b" -> 0, "c" -> 5).mapValues(_.asInstanceOf[Object]).asJava
    println("test4 " + AviatorEvaluator.execute("a/(b+c)>10", javaMap)
      + "   " + AviatorEvaluator.compile("a/(b+c)>10").execute(javaMap))
  }

  def test5(sc: SparkContext) = {
    val list = Array("hello", "world")
    val nums = Array(1, 2, 3, 4, 5)
    val map2 = scala.collection.mutable.Map("date" -> LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))).asJava
    val value = AviatorEvaluator.exec("date_to_string(dd, \"yyyy-MM-dd\")", new Date().asInstanceOf[Object])
    map2.put("date", value.toString)
    val map3 = Map("list" -> list, "nums" -> nums, "map2" -> map2)
    println("list[0]+ ' ' + list[1] = " + AviatorEvaluator.execute("list[0]+ ' ' + list[1]", map3))
    println(AviatorEvaluator.execute("'nums[0]+nums[1]+nums[2]=' + (nums[0]+nums[1]+nums[2])", map3))
    println(AviatorEvaluator.execute("'当前时间为：' + map2.date", map3))
  }

  def main(args: Array[String]): Unit = {
    // App
    // Spark框架
    val sparConf = new SparkConf()
      .setMaster("local")
      .setAppName("WordCount")
    val sc = new SparkContext(sparConf)

    // aviator 示例
    test0(sc)
    // 表达式
    test1(sc)
    // map
    test2(sc)
    // 函数
    test3(sc)
    // 编译表达式
    test4(sc)
    // 访问数组和集合
    test5(sc)
    sc.stop()
  }
}
